from pyspark.sql import SparkSession, DataFrame
import os


from cn.itcast.tag.base.BaseModel import BaseModel


"""
-------------------------------------------------
   Description :	TODO：
   SourceFile  :	GenderModel1
   Author      :	itcast team
-------------------------------------------------
"""


# 0.设置系统环境变量
# 设置 JAVA_HOME 环境变量
os.environ['JAVA_HOME'] = '/export/server/jdk1.8.0_241/'
# 设置 SPARK_HOME 环境变量
os.environ['SPARK_HOME'] = '/export/server/spark'
# 设置 PYSPARK_PYTHON 环境变量
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python'
# 设置 PYSPARK_DRIVER_PYTHON 环境变量
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python'


class GenderModel1(BaseModel):
    def compute(self, es_df, five_df):

        # 打印 es_df 的 Schema
        es_df.printSchema()
        # 显示 es_df 的数据
        es_df.show()
        # 打印 five_df 的 Schema
        five_df.printSchema()
        # 显示 five_df 的数据
        five_df.show()
        # 根据 'gender' 和 'rule' 列进行左连接，并选择 'id' 列重命名为 'userId' 和 'id' 列重命名为 'tagsId'
        new_df = es_df.join(other=five_df,
                            on=es_df['gender'] == five_df['rule'],
                            how='left').select(es_df['id'].alias("userId"), five_df['id'].alias("tagsId"))
        return new_df




if __name__ == '__main__':
    # 创建 GenderModel1 实例
    genderModel = GenderModel1(4)
    # 执行 GenderModel1 实例的 execute 方法
    genderModel.execute()